package org.example;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Hello world!
 */
public class App2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        StreamTableEnvironment tableEnvironment1 = StreamTableEnvironment.create(environment);


        String createTable = "  CREATE TABLE source_table(\n" +
                "                username STRING ,\n" +
                "                password STRING ,\n" +
                "               PRIMARY KEY(username) NOT ENFORCED\n" +
                "        )\n" +
                "        WITH(\n" +
                "                'connector' ='jdbc',\n" +
                "                'url' ='jdbc:mysql://10.0.6.42:3306/xxl_job',\n" +
                "                'username' ='root',\n" +
                "                'password' ='Rxsk@2022',\n" +
                "                'table-name' ='xxl_job_user'\n" +
                "        )";
        tableEnvironment1.executeSql(createTable);


        String sink = "  CREATE TABLE sink_table(\n" +
                "                username STRING ,\n" +
                "                password STRING ,\n" +
                "               PRIMARY KEY(username) NOT ENFORCED\n" +
                "        )\n" +
                "        WITH(\n" +
                "                'connector' ='print'\n" +
                "        )";
        tableEnvironment1.executeSql(sink);



        Table clickTable = tableEnvironment1.from("source_table");
        tableEnvironment1.createTemporaryView("clickTable",clickTable);

        String sql = "insert into sink_table select clickTable.username,clickTable.password from clickTable left join clickTable ta on clickTable.username = ta.username";

        TableResult tableResult = tableEnvironment1.executeSql(sql);


        tableResult.getJobClient().get().getJobExecutionResult().get();

        environment.execute();
    }
}